package tjs.study.notes.dotest.zookper.locker.mini;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tjs.study.notes.dotest.zookper.locker.DoTestOfMiniCurator;
import tjs.study.notes.dotest.zookper.locker.common.RLock;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * Created by Nicky.Tang on 2021/11/9 8:46 下午
 *
 * @since 02.12.10
 */
public class RLockMiniZP implements RLock {
    private static final Logger logger = LoggerFactory.getLogger(DoTestOfMiniCurator.class);

    private String key;
    private String lockPath;
    private String prePath;

    private ZkClient zkClient;
    private CountDownLatch latch;
    private Integer lockCount;

    public RLockMiniZP(String key, ZkClient zkClient) {
        this.key = key;
        this.zkClient = zkClient;
    }

    @Override
    public boolean lock() {
        try {
            // 判断是否已经加过锁了(简易可重入锁)
            if (lockCount != null) {
                lockCount++;
                return true;
            }

            // 1.尝试获取锁
            boolean hasGetLock = tryLock();
            if (!hasGetLock) {
                // 1.1 如果加锁失败，则监听上一个节点
                watchPreNode(prePath);// 这里先阻塞，不扩展尝试期限
                return lock();
            }
            // 1.2 加锁成功
            return true;
        } catch (Throwable e) {
            e.printStackTrace();
            return false;
        }
    }

    private boolean tryLock() {
        // 1. 尝试新增临时有序节点
        if (lockPath == null) {
            tryCreateParentPath();
            lockPath = zkClient.createEphemeralSequential(this.buildKey() + "/", "try lock");
        }

        String curNode = lockPath.replaceAll(this.buildKey() + "/", "");

        // 2. 判断是否为最小节点
        List<String> nodes = zkClient.getChildren(this.buildKey());
        List<String> nodesAsc = nodes.stream().sorted().collect(Collectors.toList());
        String firstNode = nodesAsc.get(0);
        // 2.1 如果是，则加锁成功，返回true
        if (curNode.equals(firstNode)) {
            lockCount = 0;
            //logger.info("{} 成功持有锁，开始执行业务。。。", this.keyPath);
            return true;
        }

        // 2.2 如果不是，则加锁失败，监听上一个节点
        this.prePath = findPreNode(curNode, nodesAsc);
        return false;
    }

    private void tryCreateParentPath() {
        if (!zkClient.exists(this.buildKey())) {
            try {
                zkClient.createPersistent(this.buildKey(), true);// todo 这里应该也用临时节点
            } catch (ZkNodeExistsException ex) {

            }
        }
    }

    private void watchPreNode(String prePath) throws InterruptedException {
        latch = new CountDownLatch(1);
        IZkDataListener perDeleteListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                if (latch != null) {
                    latch.countDown();
                }
            }
        };
        // 监听&等待
        zkClient.subscribeDataChanges(prePath, perDeleteListener);
        logger.info("{} 加锁失败，开始监听。。。", this.getLockPath());
        latch.await();

        // 释放监听
        zkClient.unsubscribeDataChanges(prePath, perDeleteListener);
    }

    private String findPreNode(String curNode, List<String> nodesAsc) {
        String preNode = null;
        for (int i = 0; i < nodesAsc.size(); i++) {
            if (curNode.equals(nodesAsc.get(i))) {
                preNode = nodesAsc.get(i - 1);
                break;
            }
        }
        return this.buildKey() + "/" + preNode;
    }

    @Override
    public void unlock() {
        if (lockCount == null) {
            return;
        }
        if (lockCount <= 0) {
            zkClient.delete(lockPath);
            //logger.info("{} 成功释放锁，开始执行业务。。。", this.lockPath);
        }
        lockCount--;
    }

    @Override
    public String getLockPath() {
        return this.lockPath;
    }

    public String buildKey() {
        return DoTestOfMiniCurator.curServerPath + this.key;
    }

    @Override
    public void close() throws Exception {
        this.unlock();
    }
}
